package cn.doitedu.udfs;

import cn.doitedu.pojo.EventStateBean;
import cn.doitedu.pojo.LogBean;
import cn.doitedu.pojo.MatchResult;
import cn.doitedu.pojo.RuleCondition;
import cn.doitedu.service.QueryService;
import cn.doitedu.utils.RuleSimulator;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @create: 2021-11-12 19:19
 * @author: 静如海的男人
 * @program: eagle-text
 **/

public class RulesMatchingKeyedProcessFunction extends KeyedProcessFunction<String, LogBean, MatchResult> {
    //匹配规则的封装类
    private QueryService queryService;
    //定义的规则
    private RuleCondition rcn;
    //状态
    private transient ListState<EventStateBean> eventListState;

    //初始化
    @Override
    public void open(Configuration parameters) throws Exception {
        ParameterTool parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();

        //定义状态描述器
        ListStateDescriptor<EventStateBean> eventStateDescriptor = new ListStateDescriptor<>("EventStateList", EventStateBean.class);
        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(2)).build();
        eventStateDescriptor.enableTimeToLive(ttlConfig);
        //初始化或恢复状态
        eventListState = getRuntimeContext().getListState(eventStateDescriptor);

        queryService = new QueryService();
        queryService.init(parameterTool,eventListState);

        rcn = RuleSimulator.getRule();

    }

    @Override
    public void processElement(LogBean bean, Context ctx, Collector<MatchResult> out) throws Exception {
        //将两个小时的数据保存到状态中
        eventListState.add(bean.toEventStateBean());

        boolean fag=queryService.isMatch(bean,rcn);

        if (fag){
            out.collect(new MatchResult(bean.getDeviceId(),bean.getTimeStamp(),System.currentTimeMillis(),rcn.getRuleId()));
        }
    }
    //释放资源
    @Override
    public void close() throws Exception {
        queryService.close();
    }
}
